package com.chencong.online.driver.dwd;

import com.chencong.online.bean.AdClickCountBean;
import com.chencong.online.bean.AdClinkBehaviorBean;
import com.chencong.online.bean.BlackListWarning;
import com.chencong.online.bean.UniqueVisitorBean;
import com.chencong.online.env.FlinkEnv;
import com.chencong.online.function.AdClinkBehaviorMapFunc;
import com.chencong.online.function.AdClinkCountAggFunc;
import com.chencong.online.function.AdClinkWindowFunc;
import com.chencong.online.function.BlackListProcessFunc;
import com.chencong.online.utils.StarRocksUtil;
import com.starrocks.connector.flink.row.StarRocksSinkRowBuilder;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.TableSchema;
import org.apache.flink.util.OutputTag;

import java.time.Duration;

/**
 * @program: user-behavior-analysis-online
 * @ClassName DwdAdAnalysisDriver
 * @description:实时广告点击分析（对频繁点击同一地区同一广告进行风控）
 * @author: chencong
 * @create: 2021-12-29 13:41
 **/
public class DwdAdAnalysisDriver {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = FlinkEnv.FlinkDataStreamRunEnv();
        env.setParallelism(1);

        //1、source
        DataStreamSource<String> inputDS = env.readTextFile("E:\\code\\user-behavior-analysis-online\\src\\main\\resources\\AdClickLog.csv");
//        inputDS.print("初始化数据");

        //2、transform
        //将数据格式化并
        SingleOutputStreamOperator<AdClinkBehaviorBean> adClinkBehaviorDS = inputDS
                .map(new AdClinkBehaviorMapFunc())
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<AdClinkBehaviorBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(new SerializableTimestampAssigner<AdClinkBehaviorBean>() {
                            @Override
                            public long extractTimestamp(AdClinkBehaviorBean element, long recordTimestamp) {
                                return element.getTimestamp() * 1000L;
                            }
                        }));
//        adClinkBehaviorDS.print("格式化数");
        //对同一个用户点击同一个广告的行为进行检测报警
        SingleOutputStreamOperator<AdClinkBehaviorBean> blackListDS = adClinkBehaviorDS
                .keyBy("userId", "adId") // 基于用户id和广告id做分组
                .process(new BlackListProcessFunc(10));
        //获取侧输出流中的报警状态
        blackListDS.getSideOutput(new OutputTag<BlackListWarning>("blackList") {
        }).print("黑名单数据");

        //基于省份分组，开窗聚合
        SingleOutputStreamOperator<AdClickCountBean> resultDS = blackListDS
                .keyBy(data -> data.getProvince())
                .window(SlidingEventTimeWindows.of(Time.hours(1), Time.minutes(5)))
                .aggregate(new AdClinkCountAggFunc(), new AdClinkWindowFunc());

        resultDS.print("结果是");
        //3、sink
        // the table structure（目标表的结果和字段）
        TableSchema tableSchema = TableSchema.builder()
                .field("province", DataTypes.VARCHAR(100))
                .field("window", DataTypes.VARCHAR(100))
                .field("user_count", DataTypes.BIGINT())
                .build();

        // set the slots with streamRowData（要输出的数据）
        StarRocksSinkRowBuilder<AdClickCountBean> starRocksSinkRowBuilder = new StarRocksSinkRowBuilder<AdClickCountBean>() {
            @Override
            public void accept(Object[] objects, AdClickCountBean adClickCountBean) {
                objects[0] = adClickCountBean.getProvince();
                objects[1] = adClickCountBean.getWindowEnd();
                objects[2] = adClickCountBean.getCount();

            }
        };
        SinkFunction<AdClickCountBean> adClickCountBeanSinkFunction = StarRocksUtil.sinkStarRocks(tableSchema, "ads_ad_analysis_hour", "ads", starRocksSinkRowBuilder);

        //sink到starRocks
        resultDS.addSink(adClickCountBeanSinkFunction);
        env.execute("DwdAdAnalysisDriver");

    }
}
